package cn.doitedu.datayi.loader

import org.apache.hadoop.fs.Path
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue, TableName}
import org.apache.hadoop.hbase.client.ConnectionFactory
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2
import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object BulkLoaderDemo {
  def main(args: Array[String]): Unit = {

    val spark:SparkSession = null;
    val data:RDD[(ImmutableBytesWritable,KeyValue)] = null

    val conf = HBaseConfiguration.create()
    conf.set("fs.defaultFS", "hdfs://doit01:8020")
    conf.set("hbase.zookeeper.quorum", "doit01,doit02,doit03")
    val job = Job.getInstance(conf)

    val conn = ConnectionFactory.createConnection(conf)
    val tableName  = TableName.valueOf("demo")
    val table = conn.getTable(tableName)
    val locator = conn.getRegionLocator(tableName)


    // 将我们自己的数据保存为HFile
    HFileOutputFormat2.configureIncrementalLoad(job, table, locator)
    data.saveAsNewAPIHadoopFile("/hfile_tmp/dws_pv_se/", classOf[ImmutableBytesWritable], classOf[KeyValue], classOf[HFileOutputFormat2], job.getConfiguration)


    // 构造一个导入hfile的工具类
    new LoadIncrementalHFiles(job.getConfiguration).doBulkLoad(new Path("/hfile_tmp/dws_pv_se/"),conn.getAdmin,table,locator)

    conn.close()
    spark.close()



  }
}
